package com.majun.produtor;

import com.majun.bean.RocketmqTransactionLog;
import com.majun.mapper.RocketmqTransactionLogMapper;
import com.majun.bean.SysUser;
import com.majun.bean.SysUserDto;
import com.majun.service.SysUserService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

import java.lang.reflect.InvocationTargetException;

/**
 * @author zy
 * @version 1.0.0
 * @date 2020/6/17
 */
@Slf4j
@RocketMQTransactionListener
public class MyTransactionMessageListener implements RocketMQLocalTransactionListener {

    private final SysUserService sysUserService;
    private final RocketmqTransactionLogMapper rocketmqTransactionLogMapper;

    public MyTransactionMessageListener(SysUserService sysUserService,
                                        RocketmqTransactionLogMapper rocketmqTransactionLogMapper) {
        this.sysUserService = sysUserService;
        this.rocketmqTransactionLogMapper = rocketmqTransactionLogMapper;
    }

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        MessageHeaders headers = message.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        SysUserDto dto = (SysUserDto) o;
        SysUser user = new SysUser();
        try {
            BeanUtils.copyProperties(dto, user);
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
        try {
            //执行本地事务
            this.sysUserService.insertWithRocketLog(user, transactionId);
            //返回本地事务执行状态为提交，发送事务消息
            log.info("本地事务正常，消息可以被发送了..");
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String transactionId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        RocketmqTransactionLog transactionLog = this.rocketmqTransactionLogMapper.selectOne(
                RocketmqTransactionLog.builder()
                        .transactionId(transactionId)
                        .build());
        if (transactionLog != null) {
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}

